package com.example.demo.flux;

import org.springframework.stereotype.Component;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketMessage;
import org.springframework.web.reactive.socket.WebSocketSession;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 实现 WebSocKet 处理类
 */
@Component
public class MyWebSocketHandler implements WebSocketHandler {
    /**
     * 创建一个线程安全的map来存聊天信息，  FluxSink 代表了发送消息的通道
     */
    public static final Map<WebSocketSession, FluxSink<WebSocketMessage>> myClients = new ConcurrentHashMap<>();


    /**
     * 实现这个接口，只需要实现一个方法，这个方法可通过 WebSocketSession 获取 Flux
     * 这个方法并不需要处理具体的数据，它面向的是Flux编程
     *
     * @param webSocketSession
     * @return
     */
    @Override
    public Mono<Void> handle(WebSocketSession webSocketSession) {
        // 1、获取连接路径：得到 WenSocket 的连接路径
        String path = webSocketSession.getHandshakeInfo().getUri().getPath();
        // 2、获取用户名：获取path路径字符串最后一个斜杠之后的内容，也就是获取聊天的用户名
        String name = path.substring(path.lastIndexOf("/") + 1);

        // 3、接收消息：     接收消息时，得到的并不是具体的消息，而是 Flux ， 这个 Flux 负责消息通信，就是个消息通道
        Flux<WebSocketMessage> sourceFlux = webSocketSession.receive();
        // map() 方法里的参数，就是 function的apply()方法，通常写成 Lambda 表达式
        Mono<Void> mono1 = sourceFlux.map(message ->
                {
                    // 这个 message 是 WebSocketMessage 类型，WebSocketMessage.getPayloadAsText()负责将消息数据转成String
                    // 获取用户发送的消息
                    String payloadAsText = message.getPayloadAsText();
                    // 返回 用户名+消息
                    String nameAndMessage = name + " ： " + payloadAsText;
                    return nameAndMessage;
                })
                // 4、实现消息广播：   把消息发给每一个用户
                // 此时的 message 已经是转换之后的 message 了，这时候是 String 类型
                .doOnNext(message ->
                {
                    // 此处做消息广播,      keySet()用于遍历map中的所有key，存在一个set集合中
                    for (WebSocketSession session : myClients.keySet()) {
                        // 通过session这个key ， 获取消息通道FluxSink
                        FluxSink<WebSocketMessage> fluxSink = myClients.get(session);
                        // 调用 fluxSink 的 next() 方法向 Flux 发送消息
                        // textMessage() 方法负责将 String 转换成 WebSocketMessage，把string类型的消息转回 WebSocketMessage
                        fluxSink.next(session.textMessage(message));
                    }
                    // .then() 方法 讲解的时候说是合并上面的消息操作，百度说是异步执行
                }).then();

        // 创建要发送消息的 outFlux
        Flux<WebSocketMessage> outFlux = Flux.create(fluxSink ->
        {
            // Flux 真正发布消息用的是Flux 底层的 fluxSink
            myClients.put(webSocketSession, fluxSink);
        });

        // 发送消息
        Mono<Void> mono2 = webSocketSession.send(outFlux);
        // 把两个mono 的消息汇总起来 再返回
        Mono<Void> allMono = Mono.zip(mono1, mono2).then();
        return allMono;
    }

}




